package com.atguigu.flink.state;

import com.atguigu.flink.function.WaterSensorMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.OperatorStateStore;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.hadoop.mapred.IFile;

import java.util.ArrayList;
import java.util.List;

/**
 * Created by Smexy on 2023/4/10
 * 
 *  状态： 是flink的一大优势。
 *  
 *  计算： 有状态的计算，当前数据在运算时，需要之前保存的一些状态(变量)一起参与运算，之后再将结果保存作为状态
 *          方便后续运算。
 *                  例如：单词统计
 *                  
 *        无状态的计算:  ETL
 *
 *  ----------------------------
 *      状态总体可以分为两大类:
 *          RawState：  原始状态。
 *                          使用的变量的类型不是Flink中提供的，都是RawState。
 *                          一旦程序故障，重启，RawState不会自动恢复，需要编程人员自己去恢复。
 *                          也需要在程序运行时，由编程人员自行备份！
 *
 *          ManagedState: Flink中定义的状态。
 *                          使用的变量的类型是Flink中提供的，都称为ManagedState(管理状态)
 *                          当前声明的这个变量，被flink自动管理(备份，故障时恢复)
 *
 *                   细分为
 *                      KeyedStream：  keyed state。
 *                                          每个Task上每种key各有各的状态，互不干扰！
 *                      不是keyedStream: operator state
 *                                          一个Task上的所有数据共用一个状态。
 *                                           需要实现CheckpointedFunction,才能声明！
 *
 *   ---------------------------
 *      当前抛出异常了，Flink尝试去恢复，但是被重启策略所抑制。
 *     JobException: Recovery is suppressed by NoRestartBackoffTimeStrategy
 *
 */
public class Demo1_State
{
    public static void main(String[] args) {
        
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(2);

        /*
            设置重启策略

             fixedDelayRestart(int restartAttempts, long delayBetweenAttempts)
                隔一段时间，再尝试重启。
                restartAttempts： 最多尝试重启多少次
                delayBetweenAttempts： 两次重启直接，间隔多久
         */
        //env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,2000));

        //开启了ck(备份)后，无数条命
        env.enableCheckpointing(2000);
        env.enableCheckpointing(5000);

                env
                   .socketTextStream("hadoop102", 8888)
                   .map(new MyMapFunction()).uid("aaa")
                   .addSink(new SinkFunction<String>()
                   {
                       @Override
                       public void invoke(String value, Context context) throws Exception {
                           if (value.contains("x")){
                               //模拟抛异常
                               throw new RuntimeException("出异常了...");
                           }
                           System.out.println(value);
                       }
                   });

        
                try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
        
    }

    //进行对每次输入的字符串进行累加拼接
    public static class MyMapFunction implements MapFunction<String,String>, CheckpointedFunction
    {

        //原始状态
        //private List<String> result=new ArrayList<>();

        //使用flink提供的状态，就当一个List用
        private ListState<String> result;

        @Override
        public String map(String value) throws Exception {
            result.add(value);
            return result.get().toString();
        }

        //周期性执行，默认200ms，可以自定义设置 。快照(备份)状态
        // 所有声明的状态会自动快照
        @Override
        public void snapshotState(FunctionSnapshotContext context) throws Exception {
            System.out.println("MyMapFunction.snapshotState");
        }

        //MyMapFunction对象，在创建时，初始化状态。 执行一次
        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {
            System.out.println("MyMapFunction.initializeState");
            //从之前备份的状态中去恢复
            //保存了所有的 OperatorState
            OperatorStateStore operatorStateStore = context.getOperatorStateStore();

           //描述状态
            ListStateDescriptor<String> listStateDescriptor = new ListStateDescriptor<>("strs", String.class);
            //按照状态的类型 和描述 从 状态的存储中获取
            result = operatorStateStore.getListState(listStateDescriptor);
        }
    }
}
